package cn.tukk.sparkproject.spark.page;

import cn.tukk.sparkproject.constant.Constants;
import cn.tukk.sparkproject.dao.IPageSplitConvertRateDAO;
import cn.tukk.sparkproject.dao.ITaskDAO;
import cn.tukk.sparkproject.dao.factory.DAOFactory;
import cn.tukk.sparkproject.domain.PageSplitConvertRate;
import cn.tukk.sparkproject.domain.Task;
import cn.tukk.sparkproject.util.DateUtils;
import cn.tukk.sparkproject.util.NumberUtils;
import cn.tukk.sparkproject.util.ParamUtils;
import cn.tukk.sparkproject.util.SparkUtils;
import com.alibaba.fastjson.JSONObject;
import groovy.lang.Tuple;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;

import java.util.*;

public class MypageConvert {
    public static void main(String[] args) {
        //1构建spark上下文
        //2构建模拟数据
        //3查询指定任务
        SparkConf conf = new SparkConf()
                .setAppName(Constants.SPARK_APP_NAME_PAGE);
        SparkUtils.setMaster(conf);
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());


        //2、生成模拟数据
        SparkUtils.mockData(sc,sqlContext);
        //3、查询任务，获取任务参数
        Long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PAGE);
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        Task task = taskDAO.findById(taskid);
        if(task==null){
            System.out.println(new Date()+":cannot find this task with id["+taskid+"].");
            return;
        }
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());

        //4、查询指定日期范围内的用户访问行为数据
        JavaRDD<Row> actionRDD = SparkUtils.getActionRDDByDateRange(sqlContext, taskParam);
        /**
         * 5、生成session,row rdd
         * 对于用户访问行为数据做个映射，将其映射为<sessionid,访问形式></sessionid,访问形式>
         */
        JavaPairRDD<String,Row> sessionidActionRow=getSessionActionRDD(actionRDD);
        sessionidActionRow=sessionidActionRow.cache();

        //第6步拿到rdd以后，考虑我们需要生成切片，那么需要拿到每个session对应的访问行为数据，进行一个聚合 按照时间排序
        JavaPairRDD<String, Iterable<Row>> groupSessionActionRDD = sessionidActionRow.groupByKey();

        //第七步  每个session单跳页面切片的生成，以及页面流的匹配，算法
        JavaPairRDD<String,Integer> pageSplitRDD=generateMatchPageSplit(sc,groupSessionActionRDD,taskParam);
        //第八步  聚合每个sessionid的页面切片流
        Map<String, Object> pageSplitPVMap = pageSplitRDD.countByKey();
        //如果指定页面流是 3,2,5,8,6   现在拿到的是pageSplitPVMap 3-2,2-5,5-8,8-6
        //实际就是计算 3页面个数， 2页面个数，5页面个数，8页面个数，6页面个数
        //获取所有起始页面的pv值
        long startPagePv=getStartPagePV(taskParam,groupSessionActionRDD);

        //第九步计算目标页面流的各个页面切片的转化率
        Map<String,Double> convertRateMap=computePageSplitConvertRate(taskParam,pageSplitPVMap,startPagePv);
        //第十步  持久化页面转换率
        persistPageConvertRate(taskid,convertRateMap);

    }

    /**
     * chijiuhua
     * @param taskid
     * @param convertRateMap
     */
    private static void persistPageConvertRate(Long taskid, Map<String, Double> convertRateMap) {
        StringBuffer sb=new StringBuffer();
        for (Map.Entry<String,Double> convertRateEntry : convertRateMap.entrySet()) {
            String pageSplit = convertRateEntry.getKey();
            Double convertRate = convertRateEntry.getValue();
            sb.append(pageSplit+"="+convertRate+"|");
        }
        String convertRate = sb.toString();
        if(convertRate.length()>1) {
            convertRate = convertRate.substring(0, convertRate.length() - 1);
        }
        PageSplitConvertRate pageSplitConvertRate = new PageSplitConvertRate();
        pageSplitConvertRate.setTaskid(taskid);
        pageSplitConvertRate.setConvertRate(convertRate);

        IPageSplitConvertRateDAO pageSplitConvertRateDAO = DAOFactory.getPageSplitConvertRateDAO();
        pageSplitConvertRateDAO.insert(pageSplitConvertRate);
    }

    /**生成页面转化率
     */
    private static Map<String, Double> computePageSplitConvertRate(JSONObject taskParam, Map<String, Object> pageSplitPVMap, long startPagePv) {
        String[] targetPages = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW).split(",");
        long lastpageSplitPV=0L;
        Map<String,Double>  convertRateMap=new HashMap<String,Double>();

        for (int i=1;i<targetPages.length;i++){
            String targetPageSplit=targetPages[i-1]+"_"+targetPages[i];
            double convertRate=0;
             long targetPageSplitPV= Long.valueOf(String.valueOf(pageSplitPVMap.get(targetPageSplit)));
            if(i==1){
                convertRate=NumberUtils.formatDouble((double)targetPageSplitPV/(double) startPagePv,2);
            }else{
                convertRate=NumberUtils.formatDouble((double)targetPageSplitPV / (double)lastpageSplitPV, 2);
            }
            convertRateMap.put(targetPageSplit,convertRate);
            lastpageSplitPV=targetPageSplitPV;
        }

        return convertRateMap;
    }

    /**
     * 获取页面流中初始页面pv
     * @param taskParam
     * @param sessionidActionRow
     * @return
     */
    private static long getStartPagePV(JSONObject taskParam, JavaPairRDD<String, Iterable<Row>>  groupSessionActionRDD) {
        String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
        final Long startPageId = Long.valueOf(targetPageFlow.split(",")[0]);
        return groupSessionActionRDD.flatMap(groupRow->{
            List<Long> list=new ArrayList<Long>();
            Iterator<Row> iterator = groupRow._2.iterator();
            while (iterator.hasNext()){
                Row row=iterator.next();
                long pageId = row.getLong(3);
                if(pageId==startPageId){
                    list.add(pageId);
                }
            }
            return list;
        }).count();

    }

    /**
     * 页面切片生成 与匹配算法
     * @param sc
     * @param groupSessionActionRDD
     * @param taskParam
     * @return
     */
    private static JavaPairRDD<String,Integer> generateMatchPageSplit(
            JavaSparkContext sc, JavaPairRDD<String,Iterable<Row>> groupSessionActionRDD, JSONObject taskParam) {
        String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
        final Broadcast<String> targetPageFlowBroadCast = sc.broadcast(targetPageFlow);
        return  groupSessionActionRDD.flatMapToPair(row->{
            //定义返回一个session 返回的list
            List<Tuple2<String,Integer>> list=new ArrayList<Tuple2<String,Integer>>();
            //获取当前session访问行为迭代器
            Iterator<Row> iterator = row._2.iterator();
            //获取使用者是定的页面流   指定页面流1,2,3，4,5,6,7
            String[] targetPages = targetPageFlowBroadCast.value().split(",");
            //拿到的sesssion访问行为，默认是乱序的，希望按照时间顺序排序  对session访问行文数据按照时间排序
            //举例，反例
            //比如 3-4-5-4-10-7  排序
           List<Row> rows = new ArrayList<>();
           while(iterator.hasNext()){
               rows.add(iterator.next());
           }
            Collections.sort(rows, new Comparator<Row>() {
                @Override
                public int compare(Row o1, Row o2) {
                    //第四列是时间 默认按照升序排序
                    return (int)(DateUtils.parseTime(o1.getString(4)).getTime()-DateUtils.parseTime(o2.getString(4)).getTime());
                }
            });
           //页面切片流生成，以及页面流的匹配
            Long lastPageId=null;
            for(Row eachdata:rows){
                long pageid = eachdata.getLong(3);
                if(lastPageId==null){
                    lastPageId=pageid;
                    continue;
                }
                //生成一个页面切片 比如说顺序是 3,5,6,7,8
                //lastPageId=3   切片是3_5
                String pageSplit=lastPageId+"_"+pageid;
                for (int i = 1; i < targetPages.length; i++) {
                    //比如指定顺序是1,3,5,7,8
                    String targetPageSplit=targetPages[i-1]+"_"+targetPages[i];
                    if(pageSplit.equals(targetPageSplit)){
                        list.add(new Tuple2<String,Integer>(pageSplit,1));
                        break;
                    }

                }
                lastPageId=pageid;
            }
            return list;
        });

    }

    private static JavaPairRDD<String,Row> getSessionActionRDD(JavaRDD<Row> actionRDD) {
        return actionRDD.mapToPair(row->{
            return new Tuple2<String,Row>(row.getString(2),row);//sessionid
        });
    }


}





















